package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/9/07
 * 通过旁路缓存方式进行维度关联的模板类
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {
    private Connection hbaseConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hbaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public T map(T obj) throws Exception {
        //根据当前流中的对象获取要关联的维度主键
        String key = getRowKey(obj);
        //先根据主键到Redis中获取维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), key);
        if (dimJsonObj != null) {
            //如果在Redis中获取到了维度数据---直接将其返回（缓存命中）
            System.out.println("~~从Redis中获取" + getTableName() + "表的" + key + "数据~~~");
        } else {
            //如果在Redis中没有获取到对应的维度数据，发送请求到Hbase中查询维度
            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, getTableName(), key, JSONObject.class);
            if (dimJsonObj != null) {
                System.out.println("~~从HBase中获取" + getTableName() + "表的" + key + "数据~~~");
                //并将从HBase中查询出来的维度放到Redis中缓存起来
                RedisUtil.writeDim(jedis, getTableName(), key, dimJsonObj);
            } else {
                System.out.println("~~~没有找到" + getTableName() + "的" + key + "维度~~~");
            }
        }
        //将维度属性补充到流中对象上
        if (dimJsonObj != null) {
            addDims(obj, dimJsonObj);
        }
        return obj;
    }
}
